package client_agent

import (
	"context"
	"fmt"
	"lark/conf"
	"lark/util"
	"strings"
	"time"

	etcdclient "github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/pkg/errors"
)

func newEtcdClient(endpoints []string) (cli *etcdclient.Client, err error) {
	cli, err = etcdclient.New(etcdclient.Config{
		Endpoints:            endpoints,
		AutoSyncInterval:     conf.EtcdSyncEndpointsTime,
		DialTimeout:          conf.EtcdDialTimeout,
		DialKeepAliveTime:    conf.EtcdDialKeepAliveTime,
		DialKeepAliveTimeout: conf.EtcdDialKeepAliveTimeout,
	})
	if err != nil {
		err = errors.WithMessage(err, "new etcd client error")
		return
	}

	return cli, nil
}

func watchEtcd(cli *etcdclient.Client, cp *ClientProxy, addressRev int64) {
	defer util.PrintPanicStack()
	defer func() {
		if cli != nil {
			cli.Close()
		}

		select {
		case <-cp.closeChan:
			return
		default:
			go cp.initService()
		}
	}()

	var err error
	var bootstrap = true
	endpoints := cli.Endpoints()

retry:
	if bootstrap {
		bootstrap = false
	} else {
		//避免过于频繁的重建
		if cli != nil {
			cli.Close()
		}

		time.Sleep(time.Second)

		if cli != nil {
			cli.Close()
		}

		//每隔十秒 ping 一下 server，如果超过3秒没有收到回复则关闭连接
		cli, err = etcdclient.New(etcdclient.Config{
			Endpoints:            endpoints,
			AutoSyncInterval:     conf.EtcdSyncEndpointsTime,
			DialTimeout:          conf.EtcdDialTimeout,
			DialKeepAliveTime:    conf.EtcdDialKeepAliveTime,
			DialKeepAliveTimeout: conf.EtcdDialKeepAliveTimeout,
		})

		if err != nil {
			logger.Errorf("new etcd client error. %s", err)
			goto retry
		}

		endpoints = cli.Endpoints()

		rev, serviceNames, addresses, workerNum, err := getEtcdServiceAddress(cli)
		if err != nil {
			logger.Errorf("get etcd service address error. %s", err)
			goto retry
		}

		cp.setServer(serviceNames, addresses, workerNum)

		addressRev = rev
	}

	addressChan := cli.Watch(context.Background(), conf.EtcdServiceAddressPrefix, etcdclient.WithPrefix(), etcdclient.WithRev(addressRev))

	patrolChan := time.Tick(time.Minute * 3)

	//将自身注册到 etcd
	resp, err := cli.Grant(context.TODO(), 5)
	if err != nil {
		logger.Errorf("grant etcd lease error, when register. %s", err)
		goto retry
	}

	ip, err := util.LocalIPByHostsFile()
	if err != nil {
		logger.Fatal(err)
	}

	_, err = cli.Put(context.TODO(), "/running_lark/address/"+ip, conf.Version, etcdclient.WithLease(resp.ID))
	if err != nil {
		logger.Errorf("put etcd  error, when register. %s", err)
		goto retry
	}

	//TODO 设置超时时间
	keepaliveResp, err := cli.KeepAlive(context.TODO(), resp.ID)
	if err != nil {
		logger.Errorf("keep etcd lease error, when register. %s", err)
		goto retry
	}

	for {
		select {
		case resp := <-addressChan:
			if resp.Err() != nil {
				logger.Errorf("watch address error. %s", resp.Err())
				goto retry
			}

			for _, ev := range resp.Events {
				if ev.Type == mvccpb.DELETE {
					serviceName, address, _, err := etcdvalue2ServiceAddress(string(ev.Kv.Key))
					if err != nil {
						logger.Error(err)
						continue
					}

					logger.Infof("etcd address delete, %s, %s", serviceName, address)

					cp.delServer(serviceName, address)
				} else if ev.Type == mvccpb.PUT {
					serviceName, address, workerNum, err := etcdvalue2ServiceAddress(string(ev.Kv.Key))
					if err != nil {
						logger.Error(err)
						continue
					}

					logger.Infof("etcd address put, %s, %s", serviceName, address)

					cp.addServer(serviceName, address, workerNum)
				} else {
					logger.Errorf("etcd events type can not hand. %s, %d", string(ev.Kv.Key), ev.Type)
				}
			}

		case resp := <-keepaliveResp:
			if resp == nil {
				logger.Errorf("etcd keep alive failed")
				goto retry
			}

		case <-patrolChan:
			ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
			_, err = cli.Get(ctx, "whatever")
			cancel()
			if err != nil {
				logger.Errorf("get whatever error. %s", err)
				goto retry
			} else {
				logger.Debug("get whatever success")
			}

		case <-cp.resetEtcdCli:
			logger.Debug(" etcd client reset")
			goto retry

		case <-cp.closeChan:
			return
		}
	}

}

func watchBackupEtcd(cli *etcdclient.Client, cp *ClientProxy, addressRev int64) {
	defer util.PrintPanicStack()
	defer func() {
		if cli != nil {
			cli.Close()
		}
	}()

	var err error
	var bootstrap = true
	endpoints := cli.Endpoints()

retry:
	if bootstrap {
		bootstrap = false
	} else {
		//避免过于频繁的重建
		if cli != nil {
			cli.Close()
		}

		time.Sleep(time.Second)

		if cli != nil {
			cli.Close()
		}

		//每隔十秒 ping 一下 server，如果超过3秒没有收到回复则关闭连接
		cli, err = etcdclient.New(etcdclient.Config{
			Endpoints:            endpoints,
			AutoSyncInterval:     conf.EtcdSyncEndpointsTime,
			DialTimeout:          conf.EtcdDialTimeout,
			DialKeepAliveTime:    conf.EtcdDialKeepAliveTime,
			DialKeepAliveTimeout: conf.EtcdDialKeepAliveTimeout,
		})

		if err != nil {
			logger.Errorf("new etcd client error. %s", err)
			goto retry
		}

		endpoints = cli.Endpoints()

		rev, serviceNames, addresses, workerNum, err := getEtcdServiceAddress(cli)
		if err != nil {
			logger.Errorf("get etcd service address error. %s", err)
			goto retry
		}

		cp.setBackupServer(serviceNames, addresses, workerNum)

		addressRev = rev
	}

	addressChan := cli.Watch(context.Background(), conf.EtcdBackupServiceAddressPrefix, etcdclient.WithPrefix(), etcdclient.WithRev(addressRev))

	for {
		select {
		case resp := <-addressChan:
			if resp.Err() != nil {
				logger.Errorf("watch address error. %s", resp.Err())
				goto retry
			}

			for _, ev := range resp.Events {
				if ev.Type == mvccpb.DELETE {
					serviceName, address, _, err := etcdvalue2ServiceAddress(string(ev.Kv.Key))
					if err != nil {
						logger.Error(err)
						continue
					}

					logger.Infof("etcd backup address delete, %s, %s", serviceName, address)

					cp.delBackupServer(serviceName, address)
				} else if ev.Type == mvccpb.PUT {
					serviceName, address, workerNum, err := etcdvalue2ServiceAddress(string(ev.Kv.Key))
					if err != nil {
						logger.Error(err)
						continue
					}

					logger.Infof("etcd backup address put, %s, %s", serviceName, address)

					cp.addBackupServer(serviceName, address, workerNum)
				} else {
					logger.Errorf("etcd backup events type can not hand. %s, %d", string(ev.Kv.Key), ev.Type)
				}
			}

		case <-cp.stopBackupChan:
			logger.Info("stop watch backup services")
			return

		case <-cp.closeChan:
			return
		}
	}

}

//获取服务的地址信息，返回数据版本号，服务名称列表，服务地址列表，错误信息
func getEtcdServiceAddress(cli *etcdclient.Client) (int64, []string, []string, []string, error) {
	logger.Info("init etcd service address begin")

	ctx, cancel := context.WithTimeout(context.Background(), conf.EtcdRequestTimeout)
	defer cancel()

	resp, err := cli.Get(ctx, conf.EtcdServiceAddressPrefix, etcdclient.WithPrefix())
	if err != nil {
		return 0, nil, nil, nil, fmt.Errorf("init etcd address error. %s", err)
	}

	var l1 []string
	var l2 []string
	var l3 []string
	for _, kv := range resp.Kvs {
		serviceName, address, workerNum, err := etcdvalue2ServiceAddress(string(kv.Key))
		if err != nil {
			logger.Errorf("init etcd service error. %s", err)
			continue
		}

		logger.Infof("init etcd service, %s, %s", serviceName, address)

		l1 = append(l1, serviceName)
		l2 = append(l2, address)
		l3 = append(l3, workerNum)
	}

	logger.Infof("init etcd service address end, rev:%d", resp.Header.Revision)
	return resp.Header.Revision, l1, l2, l3, nil
}

//返回服务名，服务地址，服务 worker 数量
func etcdvalue2ServiceAddress(s string) (string, string, string, error) {
	data := strings.SplitN(s, ".", 5)
	if len(data) != 5 {
		return "", "", "", fmt.Errorf("etcd key is wrong. %s", s)
	}

	return data[2], data[4], data[3], nil
}
